UseCases/Scaling/Demo 1 - Generate Workload.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "id": "casual-charm",
   "metadata": {},
   "source": [
    "<header style=\"padding:10px;background:#f9f9f9;border-top:3px solid #00b2b1\"><img id=\"Teradata-logo\" src=\"https://www.teradata.com/Teradata/Images/Rebrand/Teradata_logo-two_color.png\" alt=\"Teradata\" width=\"220\" align=\"right\" />\n",
    "\n",
    "<p style = 'font-size:28px;font-family:Arial;color:#E37C4D'><b>VantageCloud Lake Systems Scaling and Monitoring</b></p>\n",
    "<hr>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "rolled-laundry",
   "metadata": {},
   "source": [
    "<br>\n",
    "\n",
    "<b style = 'font-size:24px;font-family:Arial;color:#E37C4D'>Demo 1 - Generate SQL Workload</b>\n",
    "\n",
    "<p style = 'font-size:16px;font-family:Arial'>This notebook will users to create a concurrent workload agains the target system, which can be used as a baseline of activity, which allows for users to see system and performance monitoring, as well as exercise Compute Cluster scaling rules.</p>\n",
    "\n",
    "<p style = 'font-size:16px;font-family:Arial'><b>Note</b> This demonstration assumes the compute group referenced in the vars.json has a profile with scaling capabilities.  If using the default environment template, the \"CG_BusGrpA_STD\" group will have a profile \"CP_BusGrpA_STD_2_XSM\" that has scaling capabilities set up.</p>\n",
    "\n",
    "<b style = 'font-size:28px;font-family:Arial;color:#E37C4D'>Demonstration Overview</b>\n",
    "\n",
    "<p style = 'font-size:16px;font-family:Arial'>This notebook consists of three primary demonstrations</p>\n",
    "<ol style = 'font-size:16px;font-family:Arial'>\n",
    "    <li><b style = 'color:#00b2b1'>Workload Profile Setup</b> - Define the queries, concurrency, and duration of run</li>\n",
    "    <li><b style = 'color:#00b2b1'>Workload Execution</b> - Submit the workload job for parallel execution</li>\n",
    "    <li><b style = 'color:#00b2b1'>Thread monitoring and control</b> - Monitor the status of the connections, stop them if desired</li>\n",
    "    </ol>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "cubic-textbook",
   "metadata": {},
   "source": [
    "<hr>\n",
    "\n",
    "<p style = 'font-size:18px;font-family:Arial;color:#E37C4D'>Python Package Imports</p>\n",
    "\n",
    "<p style = 'font-size:16px;font-family:Arial'>\n",
    "Import the required packages including teradata client packages (teradataml and teradatasql) as well as Python multithreading utilities (concurrent.futures and threading).  Note the local python file import Concurrency_Utils.py - this has some of the custom functions created to drive the lab.</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "id": "romance-omaha",
   "metadata": {},
   "outputs": [],
   "source": [
    "import warnings\n",
    "warnings.filterwarnings('ignore')\n",
    "\n",
    "import teradatasql, logging, time, math, json\n",
    "from teradataml import *\n",
    "import getpass\n",
    "import datetime\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import concurrent.futures\n",
    "\n",
    "from time import sleep\n",
    "from random import random\n",
    "from threading import current_thread, get_ident, get_native_id, Event\n",
    "from IPython.display import display\n",
    "\n",
    "import numpy as np\n",
    "import matplotlib.pyplot as plt\n",
    "import matplotlib.animation as animation\n",
    "from IPython.display import clear_output\n",
    "%matplotlib inline\n",
    "\n",
    "from Concurrency_Utils import *\n",
    "\n",
    "# set up logging for the threads\n",
    "logging.basicConfig(format='%(asctime)s - %(message)s', \n",
    "                    filename = 'thread_status.log', \n",
    "                    filemode = 'w', \n",
    "                    level=logging.INFO)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "passive-lyric",
   "metadata": {},
   "source": [
    "<hr>\n",
    "\n",
    "<p style = 'font-size:18px;font-family:Arial;color:#E37C4D'>Connect to Vantage</p>\n",
    "\n",
    "<p style = 'font-size:16px;font-family:Arial'>Before performing any operations in Vantage, we need to read system-specific information about the users, hostnames, etc.  The below code will read in a variables file (vars.json - this has been used in prior environment setup and data engineering examples) and will connect to Vantage with this information.</p> \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "id": "regulated-damages",
   "metadata": {},
   "outputs": [],
   "source": [
    "# create a local dictionary of environment-specific variables\n",
    "\n",
    "# load vars json\n",
    "with open('../../vars.json', 'r') as f:\n",
    "    session_vars = json.load(f)\n",
    "\n",
    "# Use the \"data_engineer\" and Business compute group from the base setup\n",
    "host = session_vars['environment']['host']\n",
    "username = session_vars['hierarchy']['users']['business_users'][0]['username']\n",
    "password = session_vars['hierarchy']['users']['business_users'][0]['password']\n",
    "compute_group = session_vars['hierarchy']['users']['business_users'][0]['compute_group']\n",
    "\n",
    "conn_info = {}\n",
    "conn_info['host'] = host\n",
    "conn_info['username'] = username\n",
    "conn_info['password'] = password\n",
    "conn_info['compute_group'] = compute_group\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "hourly-funeral",
   "metadata": {},
   "source": [
    "<hr>\n",
    "<p style = 'font-size:28px;font-family:Arial;color:#E37C4D'><b>Demo 1 - Design the Query Workload</b></p>\n",
    "\n",
    "\n",
    "\n",
    "<p style = 'font-size:16px;font-family:Arial'>\n",
    "The following code will allow the user to customize a \"workload profile\" which will consist of groups of parallel threads executing a defined SQL query - each of those queries will run for a specific number of iterations or duration.\n",
    "<br><br>\n",
    "The default values here have been designed to generate a workload that will run for approximately an hour, and will be used to show activity against the system so one can run monitoring and system performance queries.\n",
    "<br><br>\n",
    "If the user wishes to customize this workload profile, either copy additional code blocks, and edit the queries, number of threads, iteration delay, and choose either iteration count or run time (in seconds).  The first two code blocks offer a \"concurrency\" and \"CPU-heavy\" queries respectively.  Comment/uncomment blocks to generate the desired workload profile.</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "id": "sound-payroll",
   "metadata": {},
   "outputs": [],
   "source": [
    "profile = []\n",
    "\n",
    "\n",
    "\n",
    "#######################################################\n",
    "######## instance definition - Concurrency ############\n",
    "\n",
    "# This query will run for approximately 2 minutes\n",
    "# reduce the row count mathematically to reduce runtime\n",
    "qry = '''\n",
    "SELECT * FROM TD_UnivariateStatistics (\n",
    "  ON (SELECT * FROM retail_sample_data.sales_transaction_line_parquet_ft SAMPLE 10000000) AS InputTable\n",
    "  USING\n",
    "  TargetColumns ('UnitSellingPriceAmt', 'UnitCostAmt')\n",
    "  Stats ('ALL')\n",
    ") AS dt;\n",
    "'''\n",
    "\n",
    "# This method call will add the above query to the workload definition - provide the number of threads\n",
    "# an iteration delay, and define the total run duration in iteration count OR time in seconds\n",
    "\n",
    "profile.extend(add_workload(qry = qry, threads = 0, delay = 2, iterations = 60))\n",
    "\n",
    "########################################################\n",
    "\n",
    "\n",
    "#######################################################\n",
    "######## instance definition - CPU ####################\n",
    "\n",
    "# This query will run for approximately 3 minutes, and consume 50% CPU on the compute clusters\n",
    "# Use minimal delay and a set duration vs. iteration count.\n",
    "qry = '''\n",
    "SELECT * FROM \n",
    "(SELECT * FROM retail_sample_data.sales_transaction_line_parquet_ft SAMPLE 100000) a\n",
    "CROSS JOIN (SELECT * FROM retail_sample_data.sales_transaction_line_parquet_ft SAMPLE 10000) b\n",
    "\n",
    "SAMPLE 100;\n",
    "'''\n",
    "\n",
    "# This method call will add the above query to the workload definition - provide the number of threads\n",
    "# an iteration delay, and define the total run duration in iteration count OR time in seconds\n",
    "profile.extend(add_workload(qry = qry, threads = 4, delay = 1, duration = 7200))\n",
    "\n",
    "########################################################\n",
    "\n",
    "\n",
    "\n",
    "#######################################################\n",
    "######## instance definition ##########################\n",
    "\n",
    "# Sample query to illustrate a different mix\n",
    "# short query, use duration in seconds\n",
    "\n",
    "qry = '''SELECT COUNT(*) FROM retail_sample_data.sales_transaction_line_parquet_ft;'''\n",
    "profile.extend(add_workload(qry = qry, threads = 3, delay = 2, duration = 7200))\n",
    "\n",
    "########################################################\n",
    "\n",
    "\n",
    "\n",
    "\n",
    "#######################################################\n",
    "######## instance definition ##########################\n",
    "\n",
    "###.......####\n",
    "\n",
    "########################################################\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "seasonal-sunday",
   "metadata": {},
   "source": [
    "<hr>\n",
    "<p style = 'font-size:28px;font-family:Arial;color:#E37C4D'><b>Demo 2- Initiate the Workload</b></p>\n",
    "\n",
    "\n",
    "\n",
    "<p style = 'font-size:16px;font-family:Arial'>\n",
    "The Python concurrent.futures module allows users to execute a function in a separate thread.  In this case, we will broadcast the function across all instances of our workload profile, and for each thread provide the query, the duration, delay, and connection information.</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "id": "dependent-lafayette",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Started Thread 4331 at 2023-08-16 12:55:35.066332, -1 iterations, Query: \"\n",
      "SELECT * FROM \n",
      "(SEL...\"\n",
      "Started Thread 4332 at 2023-08-16 12:55:35.074465, -1 iterations, Query: \"\n",
      "SELECT * FROM \n",
      "(SEL...\"\n",
      "Started Thread 4333 at 2023-08-16 12:55:35.079079, -1 iterations, Query: \"\n",
      "SELECT * FROM \n",
      "(SEL...\"\n",
      "Started Thread 4334 at 2023-08-16 12:55:35.082413, -1 iterations, Query: \"\n",
      "SELECT * FROM \n",
      "(SEL...\"\n",
      "Started Thread 4335 at 2023-08-16 12:55:35.086153, -1 iterations, Query: \"SELECT COUNT(*) FROM...\"\n",
      "Started Thread 4336 at 2023-08-16 12:55:35.089078, -1 iterations, Query: \"SELECT COUNT(*) FROM...\"\n",
      "Started Thread 4337 at 2023-08-16 12:55:35.092237, -1 iterations, Query: \"SELECT COUNT(*) FROM...\"\n"
     ]
    }
   ],
   "source": [
    "# get the total number of threads in the workload profile\n",
    "conn_info['num_cons'] = len(profile)\n",
    "    \n",
    "# create a thread pool object using concurrent.futures\n",
    "executor = concurrent.futures.ThreadPoolExecutor(max_workers = conn_info['num_cons'])\n",
    "\n",
    "# call the user function for each instance in my profile to execute them in parallel\n",
    "f = [executor.submit(run_sql, q, d, e, i, dur, conn_info) for q, d, e, i, dur in profile]\n",
    "\n",
    "# object named \"f\" is an iterable representing the threads"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "center-moses",
   "metadata": {},
   "source": [
    "<hr>\n",
    "<p style = 'font-size:28px;font-family:Arial;color:#E37C4D'><b>Demo 3 - Monitor and Control Threads</b></p>\n",
    "\n",
    "\n",
    "\n",
    "<p style = 'font-size:18px;font-family:Arial;color:#E37C4D'>3.1 - Check Status</p>\n",
    "<p style = 'font-size:16px;font-family:Arial'>\n",
    "Sample code has been provided below which allows the user to check the status of the threads.  For the sake of brevity, only high-level information about the thread state (running, finished, exception, killed) is provided in this notebook.  Detailed logging is provided in a log file named \"thread_status.log\" for detailed information on the threads, query execution status, etc.\n",
    "<br><br>\n",
    "To view the log file, either double-click on it in the file browser, or open a Terminal and cat/tail the log file.</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "id": "corrected-banana",
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "--------------------------\n",
      "Thread 1: Running\n",
      "--------------------------\n",
      "Thread 2: Running\n",
      "--------------------------\n",
      "Thread 3: Running\n",
      "--------------------------\n",
      "Thread 4: Running\n",
      "--------------------------\n",
      "Thread 5: Finished\n",
      "---Result: Thread 4335 Killed\n",
      "--------------------------\n",
      "Thread 6: Finished\n",
      "---Result: Thread 4336 Killed\n",
      "--------------------------\n",
      "Thread 7: Finished\n",
      "---Result: Thread 4337 Killed\n"
     ]
    }
   ],
   "source": [
    "i = 0\n",
    "\n",
    "# iterate over the f object, check its state\n",
    "# if not running, check the function return - failed, killed, completed gracefully\n",
    "\n",
    "for r in f:\n",
    "    i+=1\n",
    "    print('--------------------------')\n",
    "    status = 'Running' if r.running() else 'Finished'\n",
    "    print(f'Thread {str(i)}: {status}')\n",
    "    if not r.running(): print(f'---Result: {str(r.result())}')"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "occasional-exception",
   "metadata": {},
   "source": [
    "<hr>\n",
    "<p style = 'font-size:18px;font-family:Arial;color:#E37C4D'>3.2 - Manually Stop the threads</p>\n",
    "<p style = 'font-size:16px;font-family:Arial'>\n",
    "A special object \"event\" was also passed to the threads when they were initiated.  We can call a \"set()\" method on this event which will be passed to the thread, and inside the thread, we will check this status, and exit the function if so. <b>Only run this code if one wishes to stop the workload prior to the scheduled exit duration/iterations.</b>\n",
    "<br><br>\n",
    "To see this logic, open the Concurrency_Utils.py file and note the 'run_sql' function definition.</p>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "id": "least-genre",
   "metadata": {},
   "outputs": [],
   "source": [
    "# use the keyword global to modify the object in the\n",
    "# threads' namespace\n",
    "\n",
    "# global profile\n",
    "for  q, d, e, i, dur in profile:  \n",
    "    # call the set() method\n",
    "    e.set()"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "greater-seven",
   "metadata": {},
   "source": [
    "<hr>\n",
    "<p style = 'font-size:28px;font-family:Arial;color:#E37C4D'><b>Workload is now running against the system; please see Demo 2 and 3 for real-time monitoring and system performance queries</b></p>"
   ]
  },
  {
   "cell_type": "markdown",
   "id": "hydraulic-emerald",
   "metadata": {},
   "source": [
    "Copyright 2023, Teradata Corporation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "id": "appreciated-taiwan",
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  },
  "toc-showmarkdowntxt": false
 },
 "nbformat": 4,
 "nbformat_minor": 5
}